"""
---------------------------------------------------------------------
- File: service_crud.py
- Date: 2025/11/5
- Author: yangzhide
- Email: xxzjyzd@126.com
- Description: 通用CRUD基类
---------------------------------------------------------------------
"""

from typing import Any, Generic, Type, TypeVar, Dict, Union

from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
from sqlalchemy import func
from sqlmodel import SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession

# 数据表模型, 继承自SQLModel
ModelType = TypeVar("ModelType", bound=SQLModel)

# 创建数据模型, 继承自BaseModel
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
# 更新数据模型, 继承自BaseModel
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)


class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    """
    通用的CRUD基类
    参数：
    - `model`: 数据库模型
    - `create_schema`: pandantic schema
    """

    def __init__(self, session: AsyncSession, model: Type[ModelType]):
        self.session = session
        self.model = model

    async def get(self, *, id: Any) -> ModelType | None:
        """
        根据id获取单条数据, 返回数据或None
        """
        return await self.session.get(self.model, id)

    async def get_multi(self, *, skip: int = 0, limit: int = 10, filters=None) -> list[ModelType] | None:
        """
        获取多条数据,支持分页
        """
        # 查询
        query = select(self.model)
        if filters:
            query = query.where(**filters)

        # 分页
        result = await self.session.exec(query.offset(skip).limit(limit))
        return result.all()  # type: ignore

    async def get_count(self, *, filters=None):
        """
        获取数据总数
        """
        # 查询
        query = select(func.count()).select_from(self.model)
        if filters:
            query = query.where(**filters)

        # 获取数据总数
        count = await self.session.exec(query)
        return count.first()

    async def create(self, *, obj_in: CreateSchemaType) -> ModelType:
        """
        创建记录
        """
        # 根据pydantic数据模型，构建当前数据表模型实例，方便存储到数据库
        # 解决了分层架构中pydantic模型示例和数据库模型示例间数据传递和转换的问题
        db_obj = self.model.model_validate(obj_in)
        self.session.add(db_obj)
        await self.session.commit()
        await self.session.refresh(db_obj)
        return db_obj

    async def update(self, *, db_obj: ModelType, obj_in: Union[UpdateSchemaType, Dict[str, Any]]) -> ModelType:
        """
        异步更新已存在的记录
        兼容pydantic和字典数据来源
        """
        obj_data = jsonable_encoder(db_obj)  # 得到数据库中原有的所有字段和值, 并转化为字典数据
        if isinstance(obj_in, BaseModel):
            update_data = obj_in.model_dump(exclude_unset=True)  # 得到用户提交的更新字段和值, 并转化为字典数据
        else:
            update_data = obj_in

        # 遍历数据库中原有的所有字段, 如果用户提交的更新字段中存在, 则更新数据库中对应字段的值
        for field in obj_data:
            if field in update_data:
                setattr(db_obj, field, update_data[field])
        self.session.add(db_obj)
        await self.session.commit()
        await self.session.refresh(db_obj)
        return db_obj

    async def remove(self, *, id: Any) -> ModelType | None:
        """
        根据id删除记录
        """
        obj = await self.session.get(self.model, id)
        if obj:
            await self.session.delete(obj)
            await self.session.commit()
        return obj
